package que.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import que.app.AppV2;
import que.bean.TradeOrderSourceBean;
import que.util.AtguiguUtil;
import que.util.FlinkSinkUtil;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.HashMap;

import static que.common.Word.TOPIC_DWD_ORDER_SOURCE_INFO;
import static que.common.Word.TOPIC_DWD_TRAFFIC_UV_DETAIL;

/**
 * @author Naruto
 * @description: 根据订单来源统计下单总金额下单人数下单次数
 * @Class_Name que.app.dws.Dws_OrderSourceWindow
 * @Date 2022/09/03 13:53
 */
public class Dws_OrderSourceWindow extends AppV2 {

    public static void main(String[] args) {
        new Dws_OrderSourceWindow().init(
                9007,
                2,
                "Dws_OrderSourceWindow",
                TOPIC_DWD_TRAFFIC_UV_DETAIL,TOPIC_DWD_ORDER_SOURCE_INFO
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, HashMap<String, DataStreamSource<String>> streamMaps) {

        DataStream<TradeOrderSourceBean> allMstStream = parseAndUnionOne(streamMaps);

        //开窗聚合
        SingleOutputStreamOperator<TradeOrderSourceBean> result = windowAndAgg(allMstStream);

        //todo 3.写出到doris中
        writeToDoris(result);
    }

    private void writeToDoris(SingleOutputStreamOperator<TradeOrderSourceBean> result) {
        result
                .map(bean ->{
                    SerializeConfig conf = new SerializeConfig();
                    // 转成json的时候, 属性名使用下划线
                    conf.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;

                    String str = JSON.toJSONString(bean, conf);

                    return str;
                })
                .addSink(FlinkSinkUtil.getDorisSink("edu.dws_order_source_window"));

    }

    private SingleOutputStreamOperator<TradeOrderSourceBean> windowAndAgg(DataStream<TradeOrderSourceBean> allMstStream) {

        SingleOutputStreamOperator<TradeOrderSourceBean> result = allMstStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                //todo 乱序程度设置大一点，因为uv 和 uj 需经过多一层处理，数据可能会迟到，所以设置大一点
                                .<TradeOrderSourceBean>forBoundedOutOfOrderness(Duration.ofSeconds(10))
                                .withTimestampAssigner((bean, ts) -> bean.getTs())
                )
                .keyBy(bean -> bean.getSc())
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .reduce(
                        new ReduceFunction<TradeOrderSourceBean>() {
                            @Override
                            public TradeOrderSourceBean reduce(TradeOrderSourceBean o1,
                                                               TradeOrderSourceBean o2) throws Exception {
                                o1.setUvCt(o1.getUvCt() + o2.getUvCt());//独立访客数人数
                                o1.setOdCt(o1.getOdCt() + o2.getOdCt());//下单次数
                                o1.setSumAmout(o1.getSumAmout().add(o2.getSumAmout()));//下单金额
                                return o1;
                            }
                        },
                        new ProcessWindowFunction<TradeOrderSourceBean, TradeOrderSourceBean, String, TimeWindow>() {
                            @Override
                            public void process(String key,
                                                Context ctx,
                                                Iterable<TradeOrderSourceBean> it,
                                                Collector<TradeOrderSourceBean> out) throws Exception {
                                TradeOrderSourceBean bean = it.iterator().next();

                                bean.setStt(AtguiguUtil.toDateTime(ctx.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDateTime(ctx.window().getEnd()));
                                bean.setCurDate(AtguiguUtil.toDateTime(System.currentTimeMillis()));//统计日期
                                out.collect(bean);
                            }
                        }
                );
        return result;

    }

    private DataStream<TradeOrderSourceBean> parseAndUnionOne(HashMap<String, DataStreamSource<String>> streamMaps) {
        //独立访客事实表，一条数据一个用户
        SingleOutputStreamOperator<TradeOrderSourceBean> stream1 = streamMaps
                .get(TOPIC_DWD_TRAFFIC_UV_DETAIL)
                .map(str -> {
                    JSONObject json = JSON.parseObject(str);
                    JSONObject common = json.getJSONObject("common");
                    JSONObject page = json.getJSONObject("page");

                    String sc = common.getString("sc");
                    String userId = common.getString("uid");

                    Long uvCT = 1l;

                    return TradeOrderSourceBean
                            .builder()
                            .sc(sc)
                            .userId(userId)
                            .uvCt(uvCT)
                            .ouCt(0l)
                            .odCt(0l)
                            .ts(json.getLong("ts"))
                            .sumAmout(new BigDecimal(0))
                            .build();
                });

        //这边算下单次数和销售总金额 和下单人数
        SingleOutputStreamOperator<TradeOrderSourceBean> stream2 = streamMaps
                .get(TOPIC_DWD_ORDER_SOURCE_INFO)
                .map(str -> {

                    /*{
                        "id": "37490",
                            "user_id": "1746",
                            "final_amount": "22",
                            "sc": "2",
                            "ts": 1662190323,
                            "pt": "2022-09-03 07:32:07.379Z"
                    }*/
                            JSONObject json = JSON.parseObject(str);

                            return TradeOrderSourceBean
                                    .builder()
                                    .userId(json.getString("user_id"))
                                    .odCt(1l)
                                    .sumAmout(new BigDecimal(json.getString("final_amount")))
                                    .ouCt(1l)
                                    .uvCt(0l)
                                    .ts(json.getLong("ts"))
                                    .sc(json.getString("sc"))
                                    .build();
                        }
                );

        return stream1.union(stream2);

    }

}
